package org.example.timer.threadpool;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class BlockingQueue<T> {
    private int maxSize;

    private ArrayDeque<T> queue;

    private ReentrantLock lock = new ReentrantLock();

    private Condition full = lock.newCondition();

    private Condition empty = lock.newCondition();

    public BlockingQueue(int maxSize) {
        this.maxSize = maxSize;
        queue = new ArrayDeque<>();
    }

    /**
     * 添加任务，无限期等待
     *
     * @param t
     */
    public void put(T t) {
        lock.lock();
        try {
            while (queue.size() == maxSize) {
                // 满了，等待
                full.await();
            }
            queue.addLast(t);
            // 通知空的信号量，有任务了
            empty.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public <T> T get() {
        T result = null;
        lock.lock();
        try {
            while (queue.size() == 0) {
                // 空了，等待
                log.info("没有任务，开始等待");
                empty.await();
                log.info("等待结束，继续判断");
            }
            result = (T) queue.pollFirst();
            // 通知满的信号，可以继续添加了
            full.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return result;
    }

    public boolean put(T t, long delayTime, TimeUnit unit) {
        lock.lock();
        try {
            long time = unit.toNanos(delayTime);
            while (queue.size() == maxSize) {
                if (time <= 0) {
                    return false;
                }
                full.awaitNanos(time);
            }
            queue.addLast(t);
            empty.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return true;
    }

    public <T> T get(long delayTime, TimeUnit unit) {
        T result = null;
        lock.lock();
        try {
            long time = unit.toNanos(delayTime);
            while (queue.size() == 0) {
                if (time <= 0)
                    return null;
                empty.awaitNanos(time);
            }
            result = (T) queue.pollFirst();
            full.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return result;
    }

    public void tryPut(T t, ThreadPool.RejectPolicy rejectPolicy) {
        lock.lock();
        try {
            if (queue.size() >= maxSize) {
                rejectPolicy.reject(this, t);
                //没满，直接添加
            } else {
                log.info("添加任务：{}", t);
                queue.addLast(t);
                // 给empty发送信号
                empty.signalAll();
            }
        } finally {
            lock.unlock();
        }
    }
}
